//! 命令处理器
//! 
//! 这是RESP协议解析层的入口，负责：
//! 1. 持续读取网络流数据
//! 2. 解析RESP协议命令
//! 3. 分发命令到存储层
//! 4. 处理响应回写
//! 5. 支持pipeline命令处理

use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
use std::collections::VecDeque;

use crate::protocol::{NetworkStream, RespParser, ConnectionManager, RespParseError/*, PipelineManager, PipelineResponse*/};
use crate::storage::{/*Command, */CommandResult, SingleThreadCommandDispatcher};
use crate::event::{ResponseWritebackDriver, /*ResponseEvent, */NetworkCallback, ConnectionId, /*RequestId*/};

/// 命令处理器配置
#[derive(Debug, Clone)]
pub struct CommandProcessorConfig {
    /// 读取缓冲区大小
    pub read_buffer_size: usize,
    /// 连接超时时间
    pub connection_timeout: Duration,
    /// 清理任务间隔
    pub cleanup_interval: Duration,
    /// 最大pipeline命令数
    pub max_pipeline_commands: usize,
    /// 命令处理超时
    pub command_timeout: Duration,
}

impl Default for CommandProcessorConfig {
    fn default() -> Self {
        Self {
            read_buffer_size: 8192, // 8KB
            connection_timeout: Duration::from_secs(300), // 5分钟
            cleanup_interval: Duration::from_secs(60), // 1分钟清理一次
            max_pipeline_commands: 100, // 最多100个pipeline命令
            command_timeout: Duration::from_secs(30), // 30秒命令超时
        }
    }
}

/// 命令处理统计
#[derive(Debug, Clone)]
pub struct CommandProcessorStats {
    /// 总连接数
    pub total_connections: u64,
    /// 活跃连接数
    pub active_connections: usize,
    /// 总处理命令数
    pub total_commands_processed: u64,
    /// 成功处理命令数
    pub successful_commands: u64,
    /// 失败命令数
    pub failed_commands: u64,
    /// 解析错误数
    pub parse_errors: u64,
    /// 平均命令处理时间（毫秒）
    pub avg_command_time_ms: f64,
}

/// 命令处理器
/// 
/// 这是整个RESP协议处理的核心组件，负责：
/// - 管理所有网络连接
/// - 解析RESP协议数据
/// - 处理pipeline命令
/// - 与存储层和响应回写层集成
pub struct CommandProcessor {
    /// 连接管理器
    connection_manager: Arc<ConnectionManager>,
    /// 存储层命令分发器
    storage_dispatcher: Arc<Mutex<SingleThreadCommandDispatcher>>,
    /// 响应回写驱动器
    response_driver: Arc<Mutex<ResponseWritebackDriver>>,
    /// 配置
    config: CommandProcessorConfig,
    /// 是否正在运行
    running: Arc<Mutex<bool>>,
    /// 统计信息
    stats: Arc<Mutex<CommandProcessorStats>>,
}

impl CommandProcessor {
    /// 创建新的命令处理器
    pub fn new(
        storage_dispatcher: SingleThreadCommandDispatcher,
        response_driver: ResponseWritebackDriver,
        config: Option<CommandProcessorConfig>,
    ) -> Self {
        let config = config.unwrap_or_default();
        
        Self {
            connection_manager: Arc::new(ConnectionManager::new()),
            storage_dispatcher: Arc::new(Mutex::new(storage_dispatcher)),
            response_driver: Arc::new(Mutex::new(response_driver)),
            config,
            running: Arc::new(Mutex::new(false)),
            stats: Arc::new(Mutex::new(CommandProcessorStats {
                total_connections: 0,
                active_connections: 0,
                total_commands_processed: 0,
                successful_commands: 0,
                failed_commands: 0,
                parse_errors: 0,
                avg_command_time_ms: 0.0,
            })),
        }
    }

    /// 启动命令处理器
    pub fn start(&self) -> Result<(), String> {
        {
            let mut running = self.running.lock().unwrap();
            if *running {
                return Err("命令处理器已在运行中".to_string());
            }
            *running = true;
        }

        // 启动存储分发器
        self.storage_dispatcher.lock().unwrap().start();

        // 启动响应回写驱动器
        self.response_driver.lock().unwrap().start()?;

        // 设置网络回调，将响应写回到连接
        let connection_manager = Arc::clone(&self.connection_manager);
        
        let network_callback: NetworkCallback = Arc::new(move |conn_id, _req_id, response_data| {
            // 将CommandResult转换为RESP字节数据并写回连接
            if let Err(_e) = connection_manager.write_response(conn_id, response_data.as_bytes()) {
                // 写入响应失败
            }
        });

        self.response_driver.lock().unwrap().set_network_callback(network_callback);

        // 启动后台清理任务
        let _cleanup_handle = self.connection_manager.start_cleanup_task(
            self.config.cleanup_interval,
            self.config.connection_timeout,
        );

        // 命令处理器启动成功
        Ok(())
    }

    /// 停止命令处理器
    pub fn stop(&self) -> Result<(), String> {
        {
            let mut running = self.running.lock().unwrap();
            if !*running {
                return Ok(()); // 已经停止
            }
            *running = false;
        }

        // 停止响应回写驱动器
        self.response_driver.lock().unwrap().stop()?;

        // 停止存储分发器
        self.storage_dispatcher.lock().unwrap().stop();

        // 命令处理器已停止
        Ok(())
    }

    /// 处理新连接
    pub fn handle_connection(&self, stream: Box<dyn NetworkStream>) -> Result<(), String> {
        let connection_id = self.connection_manager.register_connection(stream);
        
        // 更新统计
        {
            let mut stats = self.stats.lock().unwrap();
            stats.total_connections += 1;
            stats.active_connections = self.connection_manager.get_active_connections().len();
        }

        // 启动连接处理线程
        self.spawn_connection_handler(connection_id)?;
        
        Ok(())
    }

    /// 为连接启动处理线程
    fn spawn_connection_handler(&self, connection_id: ConnectionId) -> Result<(), String> {
        let connection_manager = Arc::clone(&self.connection_manager);
        let storage_dispatcher = Arc::clone(&self.storage_dispatcher);
        let response_driver = Arc::clone(&self.response_driver);
        let running = Arc::clone(&self.running);
        let stats = Arc::clone(&self.stats);
        let config = self.config.clone();

        thread::Builder::new()
            .name(format!("conn-handler-{connection_id}"))
            .spawn(move || {
                Self::connection_handler_loop(
                    connection_id,
                    connection_manager,
                    storage_dispatcher,
                    response_driver,
                    running,
                    stats,
                    config,
                );
            })
            .map_err(|e| format!("创建连接处理线程失败: {e}"))?;

        Ok(())
    }

    /// 连接处理循环（在专用线程中运行）
    fn connection_handler_loop(
        connection_id: ConnectionId,
        connection_manager: Arc<ConnectionManager>,
        storage_dispatcher: Arc<Mutex<SingleThreadCommandDispatcher>>,
        response_driver: Arc<Mutex<ResponseWritebackDriver>>,
        running: Arc<Mutex<bool>>,
        stats: Arc<Mutex<CommandProcessorStats>>,
        config: CommandProcessorConfig,
    ) {
        // 开始处理连接
        
        let mut resp_parser = RespParser::new();
        let mut read_buffer = vec![0u8; config.read_buffer_size];
        let mut command_queue = VecDeque::new(); // Pipeline命令队列
        
        while *running.lock().unwrap() && connection_manager.is_connection_active(connection_id) {
            // 1. 从连接读取数据
            match connection_manager.read_from_connection(connection_id, &mut read_buffer) {
                Ok(0) => {
                    // 连接已关闭
                    // 连接已关闭
                    break;
                }
                Ok(bytes_read) => {
                    // 将读取的数据添加到解析器
                    resp_parser.feed_data(&read_buffer[..bytes_read]);
                }
                Err(_e) => {
                    // 从连接读取数据失败
                    break;
                }
            }

            // 2. 尝试解析命令（可能有多个pipeline命令）
            loop {
                match resp_parser.parse_value() {
                    Ok(Some(resp_value)) => {
                        // 成功解析出一个RESP值，转换为命令
                        match resp_parser.resp_to_command(resp_value) {
                            Ok(command) => {
                                command_queue.push_back(command);
                                
                                // 限制pipeline队列大小
                                if command_queue.len() > config.max_pipeline_commands {
                                    // pipeline队列已满
                                    command_queue.pop_front();
                                }
                            }
                            Err(_e) => {
                                // 命令转换失败
                                
                                // 更新错误统计
                                {
                                    let mut stats_guard = stats.lock().unwrap();
                                    stats_guard.parse_errors += 1;
                                }
                                
                                // 发送错误响应
                                let error_response = b"-ERR Protocol error\r\n";
                                let _ = connection_manager.write_response(connection_id, error_response);
                            }
                        }
                    }
                    Ok(None) => {
                        // 数据不完整，需要读取更多数据
                        break;
                    }
                    Err(RespParseError::Incomplete) => {
                        // 数据不完整，继续读取
                        break;
                    }
                    Err(_e) => {
                        // RESP解析错误
                        
                        // 更新错误统计
                        {
                            let mut stats_guard = stats.lock().unwrap();
                            stats_guard.parse_errors += 1;
                        }
                        
                        // 发送错误响应
                        let error_response = b"-ERR Protocol error\r\n";
                        let _ = connection_manager.write_response(connection_id, error_response);
                        break;
                    }
                }
            }

            // 3. 处理队列中的命令
            while let Some(command) = command_queue.pop_front() {
                let command_start = Instant::now();
                let command_name = command.name().to_string();
                
                // 生成请求ID并记录
                let request_id = connection_manager.add_pending_request(connection_id, command_name.clone());
                
                // 分发命令到存储层
                match storage_dispatcher.lock().unwrap().dispatch_command(command) {
                    Ok(()) => {
                        // 命令分发成功，等待异步处理完成
                        // 这里简化处理：直接模拟命令执行结果
                        // 实际中应该通过事件系统异步处理
                        
                        // 暂时使用同步模拟（实际应该是异步事件驱动）
                        let result = CommandResult::Ok; // 简化：所有命令都返回OK
                        
                        // 提交响应到回写驱动器
                        let response_driver_guard = response_driver.lock().unwrap();
                        match response_driver_guard.submit_command_result(
                            connection_id,
                            request_id,
                            result,
                            command_name.clone(),
                            command_start.elapsed().as_millis() as u64,
                        ) {
                            Ok(()) => {
                                // 更新成功统计
                                let mut stats_guard = stats.lock().unwrap();
                                stats_guard.total_commands_processed += 1;
                                stats_guard.successful_commands += 1;
                                
                                let elapsed_ms = command_start.elapsed().as_millis() as f64;
                                stats_guard.avg_command_time_ms = 
                                    (stats_guard.avg_command_time_ms * (stats_guard.successful_commands - 1) as f64 + elapsed_ms) 
                                    / stats_guard.successful_commands as f64;
                            }
                            Err(_e) => {
                                // 提交响应失败
                                
                                // 更新失败统计
                                let mut stats_guard = stats.lock().unwrap();
                                stats_guard.total_commands_processed += 1;
                                stats_guard.failed_commands += 1;
                            }
                        }
                    }
                    Err(e) => {
                        // 命令分发失败
                        
                        // 更新失败统计
                        {
                            let mut stats_guard = stats.lock().unwrap();
                            stats_guard.total_commands_processed += 1;
                            stats_guard.failed_commands += 1;
                        }
                        
                        // 直接发送错误响应
                        let error_response = format!("-ERR {e}\r\n");
                        let _ = connection_manager.write_response(connection_id, error_response.as_bytes());
                    }
                }
                
                // 完成请求处理
                connection_manager.complete_request(request_id);
            }

            // 4. 短暂休眠，避免CPU占用过高
            thread::sleep(Duration::from_millis(1));
        }

        // 连接处理结束，注销连接
        connection_manager.unregister_connection(connection_id);
        
        // 更新统计
        {
            let mut stats_guard = stats.lock().unwrap();
            stats_guard.active_connections = connection_manager.get_active_connections().len();
        }
        
        // 连接处理完成
    }

    /// 获取统计信息
    pub fn get_stats(&self) -> CommandProcessorStats {
        let stats = self.stats.lock().unwrap();
        let mut result = stats.clone();
        
        // 更新活跃连接数
        result.active_connections = self.connection_manager.get_active_connections().len();
        
        result
    }

    /// 打印统计信息
    pub fn print_stats(&self) {
        let stats = self.get_stats();
        
        println!("📊 命令处理器统计:");
        println!("  总连接数: {}", stats.total_connections);
        println!("  活跃连接数: {}", stats.active_connections);
        println!("  总处理命令数: {}", stats.total_commands_processed);
        println!("  成功命令数: {}", stats.successful_commands);
        println!("  失败命令数: {}", stats.failed_commands);
        println!("  解析错误数: {}", stats.parse_errors);
        println!("  平均处理时间: {:.2}ms", stats.avg_command_time_ms);
        
        if stats.total_commands_processed > 0 {
            let success_rate = stats.successful_commands as f64 / stats.total_commands_processed as f64 * 100.0;
            println!("  成功率: {success_rate:.2}%");
        }
        
        // 打印连接管理器统计
        self.connection_manager.print_stats();
        
        // 打印响应驱动器统计
        self.response_driver.lock().unwrap().print_stats();
    }

    /// 获取连接管理器引用（用于外部访问）
    pub fn connection_manager(&self) -> &Arc<ConnectionManager> {
        &self.connection_manager
    }

    /// 检查是否正在运行
    pub fn is_running(&self) -> bool {
        *self.running.lock().unwrap()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::protocol::NetworkStream;
    use crate::storage::Storage;
    use std::io;

    // 模拟网络流
    #[derive(Debug)]
    struct MockNetworkStream {
        read_data: Vec<u8>,
        write_data: Vec<u8>,
        read_pos: usize,
        connected: bool,
    }

    impl MockNetworkStream {
        fn new(data: Vec<u8>) -> Self {
            Self {
                read_data: data,
                write_data: Vec::new(),
                read_pos: 0,
                connected: true,
            }
        }

        fn written_data(&self) -> &[u8] {
            &self.write_data
        }
    }

    impl NetworkStream for MockNetworkStream {
        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
            if self.read_pos >= self.read_data.len() {
                return Ok(0);
            }
            
            let to_copy = std::cmp::min(buf.len(), self.read_data.len() - self.read_pos);
            buf[..to_copy].copy_from_slice(&self.read_data[self.read_pos..self.read_pos + to_copy]);
            self.read_pos += to_copy;
            Ok(to_copy)
        }
        
        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
            self.write_data.extend_from_slice(buf);
            Ok(buf.len())
        }
        
        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
        
        fn is_connected(&self) -> bool {
            self.connected
        }
        
        fn remote_addr(&self) -> Option<String> {
            Some("127.0.0.1:12345".to_string())
        }
        
        fn shutdown(&mut self) -> io::Result<()> {
            self.connected = false;
            Ok(())
        }
    }

    #[test]
    fn test_command_processor_lifecycle() {
        let _storage = Storage::new();
        let dispatcher = SingleThreadCommandDispatcher::new();
        let response_driver = ResponseWritebackDriver::new("test-driver", 1000);
        
        let processor = CommandProcessor::new(dispatcher, response_driver, None);
        
        // 启动处理器
        assert!(processor.start().is_ok());
        assert!(processor.is_running());
        
        // 停止处理器
        assert!(processor.stop().is_ok());
        assert!(!processor.is_running());
    }

    #[test]
    fn test_connection_handling() {
        let _storage = Storage::new();
        let dispatcher = SingleThreadCommandDispatcher::new();
        let response_driver = ResponseWritebackDriver::new("test-driver", 1000);
        
        let processor = CommandProcessor::new(dispatcher, response_driver, None);
        processor.start().unwrap();
        
        // 创建模拟连接
        let stream = Box::new(MockNetworkStream::new(
            b"*2\r\n$4\r\nPING\r\n".to_vec()
        ));
        
        // 处理连接
        assert!(processor.handle_connection(stream).is_ok());
        
        // 验证统计信息
        let stats = processor.get_stats();
        assert_eq!(stats.total_connections, 1);
        
        processor.stop().unwrap();
    }

    #[test]
    fn test_pipeline_commands() {
        // 测试pipeline命令处理
        let mut parser = RespParser::new();
        
        // 两个连续的命令
        let pipeline_data = b"*2\r\n$4\r\nPING\r\n*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n";
        parser.feed_data(pipeline_data);
        
        // 解析第一个命令
        let cmd1 = parser.parse_value().unwrap().unwrap();
        let parsed_cmd1 = parser.resp_to_command(cmd1).unwrap();
        assert_eq!(parsed_cmd1.name(), "PING");
        
        // 解析第二个命令
        let cmd2 = parser.parse_value().unwrap().unwrap();
        let parsed_cmd2 = parser.resp_to_command(cmd2).unwrap();
        assert_eq!(parsed_cmd2.name(), "SET");
    }
}